/*
 * Copyright 2020-2021 the original author(https://github.com/wj596)
 *
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */
package endpoint

import (
	"context"
	"github.com/siddontang/go-mysql/canal"
	"go-mysql-transfer/metrics"
	"go-mysql-transfer/util/helper"
	"go-mysql-transfer/util/stringutil"
	"strings"
	"sync"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"github.com/apache/rocketmq-client-go/v2/rlog"
	"github.com/juju/errors"
	"github.com/siddontang/go-mysql/mysql"

	"go-mysql-transfer/global"
	"go-mysql-transfer/model"
	"go-mysql-transfer/util/logagent"
	"go-mysql-transfer/util/logs"
)

const _rocketRetry = 2

type RocketEndpoint struct {
	client    rocketmq.Producer
	retryLock sync.Mutex
}

func newRocketEndpoint() *RocketEndpoint {
	rlog.SetLogger(logagent.NewRocketmqLoggerAgent())
	cfg := global.Cfg()

	options := make([]producer.Option, 0)
	serverList := strings.Split(cfg.RocketmqNameServers, ",")
	options = append(options, producer.WithNameServer(serverList))
	options = append(options, producer.WithRetry(_rocketRetry))
	if cfg.RocketmqGroupName != "" {
		options = append(options, producer.WithGroupName(cfg.RocketmqGroupName))
	}
	if cfg.RocketmqInstanceName != "" {
		options = append(options, producer.WithInstanceName(cfg.RocketmqInstanceName))
	}
	if cfg.RocketmqAccessKey != "" && cfg.RocketmqSecretKey != "" {
		options = append(options, producer.WithCredentials(primitive.Credentials{
			AccessKey: cfg.RocketmqAccessKey,
			SecretKey: cfg.RocketmqSecretKey,
		}))
	}

	producer, _ := rocketmq.NewProducer(options...)
	r := &RocketEndpoint{}
	r.client = producer
	return r
}

func (s *RocketEndpoint) Connect() error {
	return s.client.Start()
}

func (s *RocketEndpoint) Ping() error {
	ping := &primitive.Message{
		Topic: "BenchmarkTest",
		Body:  []byte("ping"),
	}
	_, err := s.client.SendSync(context.Background(), ping)
	return err
}

func (s *RocketEndpoint) Consume(from mysql.Position, rows []*model.RowRequest) error {
	var ms []*model.MQRespond

	var wg sync.WaitGroup

	// 一组 5个批量发送
	maxNum := 5
	rowLen := len(rows)
	for index, row := range rows {
		rule, _ := global.RuleIns(row.RuleKey)
		if rule.TableColumnSize != len(row.Row) {
			logs.Warnf("%s schema mismatching", row.RuleKey)
			continue
		}
		metrics.UpdateActionNum(row.Action, row.RuleKey)
		m, err := s.buildMessage(row, rule)
		if err != nil {
			return errors.New(errors.ErrorStack(err))
		}
		ms = append(ms, m)
		if len(ms) > maxNum || index >= (rowLen-1) {
			body, err1 := json.Marshal(ms)
			ms = ms[0:0]
			if err1 != nil {
				return err1
			}

			mArr := &primitive.Message{
				Topic: rule.RocketmqTopic,
				Body:  body,
			}
			logs.Infof("topic: %s, message: %s", rule.RocketmqTopic, string(body))
			wg.Add(1)
			var callbackErr error
			err := s.client.SendAsync(context.Background(),
				func(ctx context.Context, result *primitive.SendResult, e error) {
					if e != nil {
						callbackErr = e
					}
					wg.Done()
					logs.Infof("处理完成 %s", stringutil.ToString(result))
				}, mArr)
			if err != nil {
				return err
			}
			if callbackErr != nil {
				return err
			}
			wg.Wait()
		}
	}
	logs.Infof("处理完成 %d 条数据", len(rows))
	return nil
}

func (s *RocketEndpoint) Stock(rows []*model.RowRequest) int64 {
	//expect := true
	//var ms []*primitive.Message
	//for _, row := range rows {
	//	rule, _ := global.RuleIns(row.RuleKey)
	//	if rule.TableColumnSize != len(row.Row) {
	//		logs.Warnf("%s schema mismatching", row.RuleKey)
	//		continue
	//	}
	//	m, err := s.buildMessage(row, rule)
	//	if err != nil {
	//		logs.Errorf(errors.ErrorStack(err))
	//		expect = false
	//		break
	//	}
	//	ms = append(ms, m)
	//}
	//
	//if !expect {
	//	return 0
	//}
	//
	//if len(ms) == 0 {
	//	return 0
	//}
	//
	//var wg sync.WaitGroup
	//wg.Add(1)
	//err := s.client.SendAsync(context.Background(),
	//	func(ctx context.Context, result *primitive.SendResult, e error) {
	//		if e != nil {
	//			logs.Error(errors.ErrorStack(e))
	//			expect = false
	//		}
	//		wg.Done()
	//	}, ms...)
	//
	//if err != nil {
	//	logs.Error(errors.ErrorStack(err))
	//	return 0
	//}
	//
	//wg.Wait()
	//
	//if expect {
	//	return int64(len(ms))
	//}
	return 0
}

func (s *RocketEndpoint) buildMessage(req *model.RowRequest, rule *global.Rule) (*model.MQRespond, error) {
	kvm := rowMap(req, rule, false)

	resp := new(model.MQRespond)
	resp.Action = req.Action
	resp.Timestamp = req.Timestamp
	resp.Table = rule.Table

	if rule.ReserveRawData && canal.UpdateAction == req.Action {
		old := make(map[string]interface{}, len(rule.PaddingMap))
		old = oldRowMap(req, rule, false)

		/**
		 * 显示字段逻辑
		 */
		if !rule.ShowAllColumn {
			pkColumns := make([]string, len(rule.TableInfo.PKColumns))
			for _, index := range rule.TableInfo.PKColumns {
				pkColumns = append(pkColumns, rule.TableInfo.Columns[index].Name)
			}
			for k, _ := range old {
				exists, _ := helper.InArray(k, pkColumns)
				if old[k] == kvm[k] && !exists {
					delete(old, k)
					delete(kvm, k)
				}
			}
		}

		resp.Old = old
	}

	if rule.ValueEncoder == global.ValEncoderJson {
		resp.Data = kvm
	} else {
		resp.Data = encodeValue(rule, kvm)
	}
	//
	//body, err := json.Marshal(resp)
	//if err != nil {
	//	return nil, err
	//}

	//m := &primitive.Message{
	//	Topic: rule.RocketmqTopic,
	//	Body:  body,
	//}

	return resp, nil
}

func (s *RocketEndpoint) Close() {
	if s.client != nil {
		s.client.Shutdown()
	}
}
